package com.example.flinkcourse.lesson1.window;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * 窗口处理类
 * 实现窗口数据的聚合处理
 */
public class WindowProcessor implements org.apache.flink.streaming.api.windowing.windows.WindowFunction<String, String, String, TimeWindow> {
    private static final Logger LOG = LoggerFactory.getLogger(WindowProcessor.class);

    @Override
    public void apply(String key, TimeWindow window, Iterable<String> input, Collector<String> out) throws Exception {
        LOG.info("Processing window for key: {}, start: {}, end: {}", 
            key, window.getStart(), window.getEnd());
        
        // 收集窗口内的所有数据
        List<String> windowData = new ArrayList<>();
        for (String value : input) {
            windowData.add(value);
        }
        
        // 示例：计算窗口内的数据统计信息
        int count = windowData.size();
        String summary = String.format("Window[%s] processed %d records", key, count);
        
        LOG.info(summary);
        out.collect(summary);
    }
} 